package com.itheima.consumer.listeners;

import com.itheima.consumer.entity.Message;
import com.itheima.consumer.service.IMessageService;
import lombok.extern.slf4j.Slf4j;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.DocumentHelper;
import org.dom4j.Element;
import org.dom4j.io.SAXReader;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
public class MqListener {
    @Autowired
    private IMessageService messageService;

    private ScheduledExecutorService scheduler;

    private ExecutorService executorService;

    @PostConstruct
    public void init() {
        // 初始化定时任务
        scheduler = Executors.newScheduledThreadPool(1);
        scheduler.scheduleAtFixedRate(this::processRemainingMessages, 10, 10, TimeUnit.SECONDS);

        executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    }

    @PreDestroy
    public void cleanup() {
        // 处理程序关闭时剩余的消息
        processRemainingMessages();
        // 关闭定时任务
        if (scheduler != null && !scheduler.isShutdown()) {
            scheduler.shutdown();
        }
    }
    private synchronized void processRemainingMessages() {
        if (!msgList.isEmpty()) {
            messageService.saveBatch(msgList);
            count.addAndGet(msgList.size());
            msgList.clear();
        }
    }

    AtomicInteger count = new AtomicInteger(0);

//    List<Message> msgList = new ArrayList<>();
private final List<Message> msgList = Collections.synchronizedList(new ArrayList<>());

    @PostConstruct
    public void lissdsdfsd(){
        TimerTask task = new TimerTask() {
            @Override
            public void run() {
                log.info("每秒消费数: {}",count.getAndSet(0));
            }
        };
        Timer timer = new Timer();
        timer.schedule(task,0,1000);
    }

    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueue(String msg) {
//        executorService.submit(() -> processMessage(msg));
        processMessage(msg);
    }

    public void processMessage(String msg){
        String messageType = "";
        String messageID = "";
        //dom4j
        try{
            Document document = DocumentHelper.parseText(msg);
            Element root = document.getRootElement();
            Message message = new Message();
            for (Iterator<Element> it = root.elementIterator("MessageHead"); it.hasNext(); ) {
                Element element = it.next();
                message.setMessageType(element.element("MessageType").getText());
                message.setMessageId(Integer.valueOf(element.element("MessageID").getText()));
                byte[] bytes=element.element("MessageContent").getText().getBytes("UTF8");
                message.setMessageContent(bytes);
                //messageService.save(message);
                msgList.add(message);
            }
            if(msgList.size() >= 100){
                messageService.saveBatch(msgList);
                count.addAndGet(100);
                msgList.clear();
            }
            for (Iterator<Element> it = root.elementIterator("MessageBody"); it.hasNext();) {
                Element foo = it.next();
                // do something
            }

        }catch (Exception e){
            log.error("Exception: ", e);
        };
    }

//    public Document parse(String url) throws DocumentException {
//        //创建一个SAXReader对象
//        SAXReader reader = new SAXReader();
//        Document document = reader.read(url);
//        return document;
//    }

    @RabbitListener(queues = "work.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1 收到了 work.queue的消息：【" + msg +"】");
        Thread.sleep(20);
    }

    @RabbitListener(queues = "work.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2 收到了 work.queue的消息...... ：【" + msg +"】");
        Thread.sleep(200);
    }

    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1 收到了 fanout.queue1的消息：【" + msg +"】");
    }

    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg) throws InterruptedException {
        System.out.println("消费者2 收到了 fanout.queue2的消息：【" + msg +"】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue3",durable = "true"),
            exchange = @Exchange(name = "lyh.topic3",type = ExchangeTypes.TOPIC),
            key = {"china.#","japan.#"}
    ))
    public void listenTopicQueue3(String msg) throws InterruptedException {
        System.out.println("消费者1 收到了 topic3.queue3的消息：【" + msg +"】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue4",durable = "true"),
            exchange = @Exchange(name = "lyh.topic4",type = ExchangeTypes.TOPIC),
            key = {"#.weather","#.news"}
    ))
    public void listenTopicQueue4(String msg) throws InterruptedException {
        System.out.println("消费者2 收到了 topic.queue4的消息：【" + msg +"】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1", durable = "true"),
            exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "blue"}
    ))
    public void listenDirectQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1 收到了 direct.queue1的消息：【" + msg +"】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2", durable = "true"),
            exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg) throws InterruptedException {
        System.out.println("消费者2 收到了 direct.queue2的消息：【" + msg +"】");
    }

    @RabbitListener(queues = "topic.queue1")
    public void listenTopicQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1 收到了 topic.queue1的消息：【" + msg +"】");
    }

    @RabbitListener(queues = "topic.queue2")
    public void listenTopicQueue2(String msg) throws InterruptedException {
        System.out.println("消费者2 收到了 topic.queue2的消息：【" + msg +"】");
    }

//    @RabbitListener(queues = "object.queue")
//    public void listenObject(Map<String, Object> msg) throws InterruptedException {
//        System.out.println("消费者 收到了 object.queue的消息：【" + msg +"】");
//    }
}
